package org.example.api.tableapi.udf;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.example.api.bean.SensorReading;
import scala.Tuple2;

/**
 * @author huangqihan
 * @date 2021/3/10
 */
public class TableFunc {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 基于数据流创建表
        Table table = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp ");
        // 4
        // 4.1 table api
        Split split = new Split("_");
        tableEnv.registerFunction("split", split);
        Table res = table
                .joinLateral("split(id) as res")
                .select("id, ts, res");

        // 4.2 sql
        tableEnv.createTemporaryView("sensor", table);
        Table sqlRes = tableEnv.sqlQuery("select id, ts, res from sensor, lateral table(split(id)) as resTable(res)");

        tableEnv.toAppendStream(res, Row.class).print("res");
        tableEnv.toAppendStream(sqlRes, Row.class).print("sqlRes");

        env.execute();

    }

    public static class Split extends TableFunction<Tuple2<String, Integer>> {

        private String sep = ",";

        public Split(String sep) {
            this.sep = sep;
        }

        // 此函数必须实现
        public void eval(String str) {
            for (String s : str.split(sep)) {
                collect(new Tuple2<String, Integer>(s, s.length()));
            }
        }
    }
}
